package kafka

import (
	"context"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/gogf/gf/frame/g"
	"strings"
)

type KafkaClient struct {
	Addrs 		[]string
	KafkaConfig *sarama.Config
}

func NewClient(name string) *KafkaClient {
	config := g.Cfg().GetMapStrStr(name)
	if config == nil {
		panic("初始化失败：kafka配置文件有误")
	}

	kc := &KafkaClient{}
	kc.Addrs = strings.Split(config["brokers"], ",")
	kc.KafkaConfig = sarama.NewConfig()

	// Kafka broker version is mandatory for API compatability
	version, err := sarama.ParseKafkaVersion(config["version"])
	if err != nil {
		panic(fmt.Sprintf("error parsing Kafka version: %v" + err.Error(), config["version"]))
	}
	kc.KafkaConfig.Version = version
	return kc
}

// 管理端
func (c *KafkaClient) KAdmin() sarama.ClusterAdmin {
	admin, err := sarama.NewClusterAdmin(c.Addrs, c.KafkaConfig)
	if err != nil {
		panic(err)
	}

	return admin
}

// 客户端
func (c *KafkaClient) KClient() sarama.Client {
	client, err := sarama.NewClient(c.Addrs, c.KafkaConfig)
	if err != nil {
		panic(err)
	}

	return client
}

// 同步消息生产者
func (c *KafkaClient) KSyncProducer() *kafkaSyncProducer {
	configureSyncProducer(c.KafkaConfig)
	sp, err := NewSyncProducer(c.Addrs, c.KafkaConfig)
	if err != nil {
		panic(err)
	}

	return sp
}

// 异步消息生产者
func (c *KafkaClient) KAsyncProducer(ctx context.Context) *kafkaAsyncProducer {
	configureAsyncProducer(c.KafkaConfig)
	sp, err := NewAsyncProducer(c.Addrs, c.KafkaConfig, ctx)
	if err != nil {
		panic(err)
	}

	return sp
}

// 消息消费者
func (c *KafkaClient) KConsumer(group string, handler Handler, ctx context.Context) Consumer {
	configureConsumer(c.KafkaConfig)
	consumer, err := NewConsumer(c.Addrs, c.KafkaConfig, group, handler, ctx)
	if err != nil {
		panic(err)
	}

	return consumer
}

// 消息分区消费者
func (c *KafkaClient) KPartitionConsumer(topic string, partition int32, offset int64) sarama.PartitionConsumer {
	configurePartitionConsumer(c.KafkaConfig)
	pc, err := NewPartitionConsumer(c.Addrs, c.KafkaConfig, topic, partition, offset)
	if err != nil {
		panic(err)
	}

	return pc
}

